package org.cliff.message.platform.core.service;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.cliff.message.platform.core.domain.*;
import org.cliff.message.platform.core.domain.*;
import org.cliff.message.platform.core.exception.MPBusinessException;
import org.cliff.message.platform.core.mapper.ReceivedMessageMapper;
import org.cliff.message.platform.core.mapper.ReceivedMessageResendDetailMapper;
import com.xfvape.uid.UidGenerator;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 接收消息业务处理类
 * 
 * @author 刘文
 * @date 2023/10/17
 */
@Service
public class ReceivedMessageService extends ServiceImpl<ReceivedMessageMapper, ReceivedMessage> {

    @Autowired
    private ReceivedMessageMapper mapper;

    @Autowired
    private ReceivedMessageResendDetailMapper resendDetailMapper;

    @Autowired
    private UidGenerator uidGenerator;

    /**
     * 标记消息重发
     * 
     * @param messageIds
     * @param delayTimeInMillis
     */
    @Transactional(rollbackFor = Throwable.class)
    public void markMessagesResend(List<Long> messageIds, long delayTimeInMillis) {
        List<ReceivedMessage> originalMessages = mapper.selectBatchIds(messageIds);
        if (originalMessages.isEmpty()) {
            throw new MPBusinessException("所选消息不存在");
        }

        // 查询重发状态信息
        LambdaQueryWrapper<ReceivedMessage> queryWrapper = Wrappers.lambdaQuery(ReceivedMessage.class);
        queryWrapper.in(ReceivedMessage::getOriginalMessageId, messageIds);
        List<ReceivedMessage> resendMessages = mapper.selectList(queryWrapper);

        validateMarkedMessage(originalMessages, false);
        validateMarkedMessage(resendMessages, true);

        // 生成重发消息
        long planedSendTime = System.currentTimeMillis() + delayTimeInMillis;
        List<ReceivedMessage> copiedMessage = new ArrayList<>(originalMessages.size());
        List<ReceivedMessageResendDetail> resendDetails = new ArrayList<>(originalMessages.size());
        for (ReceivedMessage originalMessage : originalMessages) {
            ReceivedMessage copy = new ReceivedMessage();
            BeanUtils.copyProperties(originalMessage, copy);
            copy.setId(uidGenerator.getUID());
            copy.setOriginalMessageId(originalMessage.getId());
            copy.setProcessStatus(ProcessStatusConstants.PENDING);
            copy.setCreateTime(null);
            copy.setUpdateTime(null);
            mapper.insert(copy);
            ReceivedMessageResendDetail resendDetail = new ReceivedMessageResendDetail();
            resendDetail.setMessageId(copy.getId());
            resendDetail.setPlanedSendTime(planedSendTime);
            resendDetail.setSendStatus(SendStatusConstants.PENDING);
            resendDetail.setSendType(SendTypeConstants.MARKED_RESEND);
            resendDetailMapper.insert(resendDetail);
        }
    }

    private void validateMarkedMessage(List<ReceivedMessage> receivedMessages, boolean checkResendMessages) {

        String runningMessages = receivedMessages.stream()
            .filter(receivedMessage -> ProcessStatusConstants.RUNNING.equals(receivedMessage.getProcessStatus())
                || ProcessStatusConstants.PENDING.equals(receivedMessage.getProcessStatus()))
            .map(receivedMessage -> checkResendMessages ? receivedMessage.getOriginalMessageId().toString()
                : receivedMessage.getId().toString())
            .collect(Collectors.joining(","));
        if (!runningMessages.isEmpty()) {
            throw new MPBusinessException(
                "消息" + runningMessages + (checkResendMessages ? "整在重发处理中，不能重发标记" : "还在处理中，不能标记重发"));
        }
    }
}
